package com.example.xw;

import com.example.xw.util.FileUtil;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.File;
import java.io.IOException;
import java.util.Properties;

/**
 * @author xiongwei_vendor
 * @date 2020/8/20 17:43
 */
public class KafkaProduct {

    String address = "10.4.192.40:10209";
    String value = "测试一下";
    String keySerializer = "StringSerializer";
    String valueSerializer = "StringSerializer";
    String topic = "iis_test";
    Long sendCount = 0L;

    public void fileData() {
        try {
            String filePath = System.getProperty("user.dir");
            File file = new File(filePath + File.separator + "config.txt");
            String fileMsg = FileUtil.readFileToString(file, "utf-8");
            String[] splMsgs = fileMsg.split("\n");
            for (int i = 0; i < splMsgs.length; i++) {
                System.out.println(" i = " + i + "  " + splMsgs[i]);
            }
            if (splMsgs.length >= 7) {
                keySerializer = splMsgs[1].replaceAll("\\r\\n|\\r|\\n", "");
                valueSerializer = splMsgs[3].replaceAll("\\r\\n|\\r|\\n", "");
                address = splMsgs[5].replaceAll("\\r\\n|\\r|\\n", "");
                topic = splMsgs[7].replaceAll("\\r\\n|\\r|\\n", "");
                String count = splMsgs[9].replaceAll("\\r\\n|\\r|\\n", "");
                sendCount = Long.parseLong(count);
            } else {
                System.out.println("读取文件config.txt失败");
            }
            File valueFile = new File(filePath + File.separator + "value.txt");
            String valueMsg = FileUtil.readFileToString(valueFile, "utf-8");
            value = valueMsg.replaceAll("\\r\\n|\\r|\\n", "");
            System.out.println("发送的数据为：" + value);
        } catch (IOException ioException) {
            ioException.printStackTrace();
        }
    }


    public void sendData() {
        //TODO 生产者三个属性必须指定(broker地址清单、key和value的序列化器)
        Properties properties = new Properties();
        properties.put("bootstrap.servers", address);
        if ("StringSerializer".equals(keySerializer)) {
            properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        } else {
            properties.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
        }
        if ("StringSerializer".equals(valueSerializer)) {
            properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        } else {
            properties.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
        }
//        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        KafkaProducer producer = new KafkaProducer(properties);
        try {
            ProducerRecord record;
            try {
                //TODO 发送 *条消息
                if (sendCount == null || sendCount < 1L) {
                    sendCount = 1L;
                }
                for (int i = 0; i < sendCount; i++) {
                    record = new ProducerRecord(topic, "", value);
                    producer.send(record); //发送并发忘记（重试会有）
                    System.out.println("===========================================" + i + "，message is sent");
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        } finally {
            producer.close();
        }
    }

}
